package com.icreate.storm.bolt;

import com.icreate.entity.indicator.IndicatorID;
import com.icreate.storm.methods.indicators.*;
import com.mongodb.client.model.Filters;
import org.apache.storm.mongodb.common.QueryFilterCreator;
import org.apache.storm.topology.BoltDeclarer;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import java.util.*;

/**
 * Created by hadoop on 17-5-3.
 */
public class IndicatorBoltBuilder {
    //            private static final String mongoURL = "mongodb://192.168.7.73:27017/test";
//    private static final String mongoURL = "mongodb://hadoop-slave1.icreate.com:27017/test";
    private String mongoURL;
    private TopologyBuilder builder;
    //    private BoltDeclarer mongoCellImsiBolt;
    private BoltDeclarer mongoCellBolt;
    private BoltDeclarer mongoImsiBolt;
    private BoltDeclarer mongoImeiBolt;
    private BoltDeclarer mongoDateBolt;

    public IndicatorBoltBuilder(TopologyBuilder builder, String mongoURL) {
        this.builder = builder;
        this.mongoURL = mongoURL;
//        GetId cellImsi_id = id -> id.getDt() + "-" + t.getValueByField("cellid") + "-" + t.getValueByField("imsi");
        GetId cell_id = t -> {
            IndicatorID id = (IndicatorID) t.getValueByField("IndicatorID");
            return id.getDt() + "-" + id.getCellid();
        };
        GetId imsi_id = t -> {
            IndicatorID id = (IndicatorID) t.getValueByField("IndicatorID");
            return id.getDt() + "-" + id.getImsi();
        };
        GetId imei_id = t -> {
            IndicatorID id = (IndicatorID) t.getValueByField("IndicatorID");
            return id.getDt() + "-" + id.getImei();
        };
        GetId date_id = t -> {
            IndicatorID id = (IndicatorID) t.getValueByField("IndicatorID");
            return id.getDt();
        };
//        QueryFilterCreator filterCellImsi = t -> {
//            List<Bson> bsons = new LinkedList<>();
//            bsons.add(Filters.eq("_id", cellImsi_id.getId(t)));
//            bsons.add(Filters.eq("dt", t.getValueByField("dt")));
//            bsons.add(Filters.eq("cellid", t.getValueByField("cellid")));
//            bsons.add(Filters.eq("imsi", t.getValueByField("imsi")));
//            return Filters.and(bsons);
//        };
        QueryFilterCreator filterCell = t ->
                Filters.eq("_id", cell_id.getId(t));
        QueryFilterCreator filterImsi = t ->
                Filters.eq("_id", imsi_id.getId(t));
        QueryFilterCreator filterImei = t ->
                Filters.eq("_id", imei_id.getId(t));
        QueryFilterCreator filterDate = t ->
                Filters.eq("_id", date_id.getId(t));
//        this.mongoCellImsiBolt = builder.setBolt("mongo-cellimsi-bolt", newMongoINCBolt("indicator_cell_imsi", filterCellImsi, cellImsi_id), 2);
        this.mongoCellBolt = builder.setBolt("mongo-cell-bolt", newMongoINCBolt("indicator_cell", filterCell, cell_id), 4);
        this.mongoImsiBolt = builder.setBolt("mongo-imsi-bolt", newMongoINCBolt("indicator_imsi", filterImsi, imsi_id), 3);
        this.mongoImeiBolt = builder.setBolt("mongo-imei-bolt", newMongoINCBolt("indicator_imei", filterImei, imei_id), 3);
        this.mongoDateBolt = builder.setBolt("mongo-date-bolt", newMongoINCBolt("indicator_date", filterDate, date_id), 2);
    }

    public void setUE_MR() {

        String boltId = "UE_MRImsiIndicatorBolt";
        builder.setBolt(boltId, new IndicatorBolt(new UE_MRIndicator()), 1).localOrShuffleGrouping("CITY_RC_MM-type-bolt", "UE_MR");
        this.setMongoBolt(boltId);
    }

    public void setUu() {
        String boltId = "UuIndicatorBolt";
        builder.setBolt(boltId, new IndicatorBolt(new UuIndicator()), 1).localOrShuffleGrouping("CITY_RC_MM-type-bolt", "Uu");
        this.setMongoBolt(boltId);
    }

    public void setX2() {
        String boltId = "X2IndicatorBolt";
        builder.setBolt(boltId, new IndicatorBolt(new X2Indicator()), 1).localOrShuffleGrouping("CITY_RC_MM-type-bolt", "X2_XDR");
        this.setMongoBolt(boltId);
    }

    public void setS1_MME() {
        String boltId = "S1_MMEIndicatorBolt";
        builder.setBolt(boltId, new IndicatorBolt(new S1_MMEIndicator()), 1).localOrShuffleGrouping("CITY_SHARE_MM-type-bolt", "MLTE_S1_MME");
        this.setMongoBolt(boltId);
    }

    public void setS1_U() {
        String boltId = "S1_UIndicatorBolt";
        builder.setBolt(boltId, new IndicatorBolt(new S1_UIndicator()), 1).localOrShuffleGrouping("CITY_SHARE_LTE_UP_MM-type-bolt", "MLTE_S1U_HTTP");

        this.setMongoBolt(boltId);
    }

    private void setMongoBolt(String boltId) {
        mongoCellBolt.fieldsGrouping(boltId, new Fields("cellid"));
        mongoImsiBolt.fieldsGrouping(boltId, new Fields("imsi"));
        mongoImeiBolt.fieldsGrouping(boltId, new Fields("imei"));
        mongoDateBolt.shuffleGrouping(boltId);
    }

    private IRichBolt newMongoINCBolt(String collectionName, QueryFilterCreator updateQueryCreator, GetId getid) {
        String[] excludeFields = {"type", "cellid", "imsi", "imei", "IndicatorID"};
        Set<String> excludeFields_set = new LinkedHashSet<>();
        for (String excludeField : excludeFields) {
            excludeFields_set.add(excludeField);
        }
        Map<String, String[]> indicatorFields = new LinkedHashMap<>();
        indicatorFields.put(S1_MMEIndicator.typeName, S1_MMEIndicator.baseFields);
        indicatorFields.put(S1_UIndicator.typeName, S1_UIndicator.baseFields);
        indicatorFields.put(UE_MRIndicator.typeName, UE_MRIndicator.baseFields);
        indicatorFields.put(UuIndicator.typeName, UuIndicator.baseFields);
        indicatorFields.put(X2Indicator.typeName, X2Indicator.baseFields);

//        MongoMapper mapper = new HourINCMongoMapper("dt", "cellid", "imsi", "hour");
//        QueryFilterCreator updateQueryCreator = new IndicatorQueryFilterCreator().withField(ids);
        MongoBufferUpdateBolt updateBolt = new MongoBufferUpdateBolt(mongoURL, collectionName, updateQueryCreator, indicatorFields, getid)
                .withUpsert(true);
        return updateBolt;
    }
}
